package com.sun.kafka.controller;

import com.alibaba.fastjson2.JSONObject;
import com.sun.kafka.producer.KafkaProducer;
import lombok.RequiredArgsConstructor;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.Date;
import java.util.Properties;

@RestController
@RequestMapping("/kafka")
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class SendController {
    public final static String TOPICS = "simple1,simple2,batch1,batch2";

    private final KafkaProducer producer;

    // http://localhost:8080/kafka/send/1
    @GetMapping(value = "/send/{num}")
    public String send(@PathVariable Integer num) {
        producer.send(num);
        return "success";
    }

    // 生成测试数据
    public static void main(String[] args) {
        producerSend();
    }

    public static void producerSend() {
        Producer<String, String> producer = getProducer();
        for (int i = 0; i < 2; i++) {
            for (String topic : TOPICS.split(",")) {
                ProducerRecord<String, String> record = new ProducerRecord<>(topic, null, buildValue(topic, i));
                producer.send(record);
            }
        }
        producer.close();
    }

    private static String buildValue(String topic, int i) {
        JSONObject json = new JSONObject();
        if (StringUtils.equals("batch1", topic)) {
            json.put("name", "metric-" + topic);
            json.put("ts", new Date());
            JSONObject labelJson = new JSONObject();
            labelJson.put("tagKey", i);
            json.put("labels", labelJson);
            JSONObject fieldJson = new JSONObject();
            fieldJson.put("valueKey", i);
            json.put("fields", fieldJson);
        } else if(StringUtils.equals("batch2", topic)) {
            json.put("name2", "metric-" + topic);
            json.put("ts2", new Date());

            JSONObject labelJson = new JSONObject();
            labelJson.put("tagKey", i);
            json.put("labels2", labelJson);

            JSONObject fieldJson = new JSONObject();
            fieldJson.put("valueKey", i);
            json.put("fields2", fieldJson);
        } else {
            json.put("key" + i, "value-" + i);
        }
        return json.toJSONString();
    }

    private static Producer<String, String> getProducer() {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.setProperty(ProducerConfig.ACKS_CONFIG, "1");
        properties.setProperty(ProducerConfig.RETRIES_CONFIG, "0");
        properties.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, "16384");
        properties.setProperty(ProducerConfig.LINGER_MS_CONFIG, "1");
        properties.setProperty(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");

        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        return new org.apache.kafka.clients.producer.KafkaProducer<>(properties);
    }

}
